package com.example.tableSql

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.table.api.{Table, TableResult}
import org.apache.flink.table.api.bridge.scala.BatchTableEnvironment
// 不能放在main方法中
case class Suppile(name:String,number:String,part:String,cost:String,date:String)

object Demo {
  def main(args: Array[String]): Unit = {

    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    val df: DataSet[Suppile] = env.readCsvFile[Suppile]("E:\\note\\flink-project\\temp\\supplier_data.csv")
    df.print()
//
    val tEnv = BatchTableEnvironment.create(env)

//    示例一
    val tb: Table = tEnv.fromDataSet(df)
    tb.printSchema()
    val table = tb.select("name, part").filter("part == '7009'")
    table.execute().print()

//    示例二
    tEnv.createTemporaryView("xxx", df)
    tEnv.executeSql("select * from xxx where part='2341'").print()
//   示例三
    tEnv.registerTable("bbb", tEnv.fromDataSet(df))
    tEnv.sqlQuery("select * from xxx where part='2341'").execute().print()
  }

}
